package com.abyss.sink.stream;


import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;

/**
 * 数据写出到Mysql中
 */
public class SinkToMysqlDemo {
    public static void main(String[] args) throws Exception {
        // 1. Env
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 2. Source UserInfo javabean对象
        DataStreamSource<UserInfo> source = env.fromElements(
                new UserInfo(1, "xiaoxiao", "123", "潇潇"),
                new UserInfo(3, "dachui", "123", "王大锤"),
                new UserInfo(5, "luozhixiang", "123", "罗志祥"),
                new UserInfo(7, "huanglei", "123", "黄磊"),
                new UserInfo(9, "zhangyixing", "123", "张艺兴")
        );

        // 3. 构建Sink
        source.addSink(new MyMysqlSink());

        // 4. 执行
        env.execute();
    }

    /*
    自定义Mysql SinkFunction
     */
    public static class MyMysqlSink extends RichSinkFunction<UserInfo> {
        private Connection connection = null;
        private PreparedStatement ps = null;

        @Override
        public void open(Configuration parameters) throws Exception {
            super.open(parameters);
            Class.forName("com.mysql.jdbc.Driver");
            String url = "jdbc:mysql://node1:3306/flinkdemo?useUnicode=true&characterEncoding=utf-8&useSSL=false";
            this.connection = DriverManager.getConnection(url, "root", "123456");
            this.ps = this.connection.prepareStatement("INSERT INTO user VALUES(?, ?, ?, ?)");
        }

        @Override
        public void close() throws Exception {
            super.close();
            if (this.ps != null) this.ps.close();
            if (this.connection != null) this.connection.close();
        }

        @Override
        public void invoke(UserInfo value, Context context) throws Exception {
            ps.setInt(1, value.getId());
            ps.setString(2, value.getUsername());
            ps.setString(3, value.getPassword());
            ps.setString(4, value.getName());

            ps.execute();
//            this.connection.commit();
        }
    }

    public static class UserInfo {
        private int id;
        private String username;
        private String password;
        private String name;

        @Override
        public String toString() {
            return "UserInfo{" +
                    "id=" + id +
                    ", username='" + username + '\'' +
                    ", password='" + password + '\'' +
                    ", name='" + name + '\'' +
                    '}';
        }

        public int getId() {
            return id;
        }

        public void setId(int id) {
            this.id = id;
        }

        public String getUsername() {
            return username;
        }

        public void setUsername(String username) {
            this.username = username;
        }

        public String getPassword() {
            return password;
        }

        public void setPassword(String password) {
            this.password = password;
        }

        public String getName() {
            return name;
        }

        public void setName(String name) {
            this.name = name;
        }

        public UserInfo() {
        }

        public UserInfo(int id, String username, String password, String name) {
            this.id = id;
            this.username = username;
            this.password = password;
            this.name = name;
        }
    }
}